package com.hlz.flink.chapter05;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * 自定义数据源
 *
 * @author Hongliang Zhu
 * @create 2023-01-02 20:43
 */
public class ClickSource implements SourceFunction<Event> {
    // 声明一个布尔变量，作为控制数据生成的标识位
    private Boolean running = true;

    /**
     * 使用运行时上下文对象（SourceContext）向下游发送数据
     *
     * @param ctx 上下文对象
     * @throws Exception
     */
    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        Random random = new Random(); // 在指定的数据集中随机选取数据
        String[] users = {"Mary", "Alice", "Bob", "Cary"};
        String[] urls = {"./home", "./cart", "./fav", "./prod?id=1",
                "./prod?id=2"};

        while (running) {
            ctx.collect(new Event(users[random.nextInt(users.length)], urls[random.nextInt(urls.length)], Calendar.getInstance().getTimeInMillis()));
            TimeUnit.SECONDS.sleep(1);
        }

    }


    /**
     * 通过标识位控制退出循环，来达到中断数据源的效果。
     */
    @Override
    public void cancel() {

        running = false;

    }
}
